package bolts;

import java.util.Date;
import java.util.List;
import java.util.Map;

import output.TopUser;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.tuple.Tuple;

public class OutputTopUsersBolt extends AbstractOutputToFileBolt {

    private final String outputPrefix;

    public OutputTopUsersBolt(String outputPrefix) {
        super();
        this.outputPrefix = outputPrefix;
    }

    @Override
    public void execute(Tuple input, BasicOutputCollector collector) {
        writer.println("Top Users at " + dateFormat.format(new Date()));

        final List<TopUser> topUsers = (List<TopUser>) input.getValueByField("topUsers");
        for (final TopUser user : topUsers) {
            writer.println(user.getUser() + " " + user.getTweetCount());
        }
        writer.flush();
    }

    @Override
    protected String getOutputFileName(Map stormConfig) {
        return this.outputPrefix + (String) stormConfig.get("topUsersOutputFile");
    }

}
